package com.rui.study.micro.scheduler.mq;

import com.rui.study.commons.util.CronUtil;
import com.rui.study.commons.util.JsonUtil;
import com.rui.study.micro.scheduler.model.Task;
import com.rui.study.micro.scheduler.service.BulletinService;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.core.ExchangeTypes;
import org.springframework.amqp.rabbit.annotation.Exchange;
import org.springframework.amqp.rabbit.annotation.Queue;
import org.springframework.amqp.rabbit.annotation.QueueBinding;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.cloud.context.config.annotation.RefreshScope;
import org.springframework.data.redis.core.HashOperations;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.io.Serializable;
import java.util.Date;

import static com.rui.study.micro.scheduler.util.Constant.BULLETIN_TASK;

/**
 * @program: study_20180903
 * @description:
 * @author: Yaowr
 * @create: 2018-11-01 11:54
 **/
@Component
@RefreshScope
public class BulletinMQListener {

    private static final Logger LOGGER = LoggerFactory.getLogger(BulletinMQListener.class);

    @Autowired
    private BulletinService bulletinService;

    @Resource(name = "redisTemplate")
    private HashOperations<Serializable, String, String> redisOps;


    @RabbitListener(
            bindings = @QueueBinding(
                value = @Queue(autoDelete = "true"),
                exchange = @Exchange(value = "${bulletin.publish.exchange}", type = ExchangeTypes.FANOUT, autoDelete = "true"),
                key = ""),
            concurrency = "1")
    public void publish(byte[] data) {
        /* 1.解析数据 */
        String json = new String(data);
        LOGGER.info("Received a bulletin message [{}]", json);
        Task task = parse(json);

        if (task != null) {
            /* 2.将任务id缓存到Redis */
            redisOps.put(BULLETIN_TASK, task.getTaskId(), json);

            /* 3.构建任务调度器， */
            bulletinService.buildScheduler(task);
        }
    }


    @RabbitListener(
            bindings = @QueueBinding(
                value = @Queue(autoDelete = "true"),
                exchange = @Exchange(value = "${bulletin.cancel.exchange}", type = ExchangeTypes.FANOUT, autoDelete = "true"),
                key = ""),
            concurrency = "1")
    public void cancel(byte[] data) {
        String taskId = new String(data);
        bulletinService.clearTask(taskId);
    }


    private Task parse(String json) {
        Task task = null;
        try {
            task = JsonUtil.toObject(json, Task.class);
            String cron = task.getCron();
            if (cron == null || cron.length() <= 0) {
                cron = CronUtil.getCron(new Date(System.currentTimeMillis() + 30000));
                task.setCron(cron);
            }
        } catch (Exception e) {
            LOGGER.error("Parse json message [{}] failed, caused by:[{}]", json, e.getMessage());
        }
        return task;
    }

}
